package com.atguigu.tingshu.album.service.impl;

import com.atguigu.tingshu.album.mapper.AlbumAttributeValueMapper;
import com.atguigu.tingshu.album.mapper.AlbumInfoMapper;
import com.atguigu.tingshu.album.mapper.AlbumStatMapper;
import com.atguigu.tingshu.album.service.AlbumAttributeValueService;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.common.cache.GuiGuCache;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.query.album.AlbumInfoQuery;
import com.atguigu.tingshu.vo.album.AlbumAttributeValueVo;
import com.atguigu.tingshu.vo.album.AlbumInfoVo;
import com.atguigu.tingshu.vo.album.AlbumListVo;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.Nullable;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.Arrays;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

    @Autowired
    private AlbumInfoMapper albumInfoMapper;

    @Autowired
    private KafkaService kafkaService;

    @Autowired
    private AlbumAttributeValueMapper albumAttributeValueMapper;

    @Autowired
    private AlbumStatMapper albumStatMapper;

    @Autowired
    private AlbumAttributeValueService albumAttributeValueService;

    @Autowired
    private RedisTemplate redisTemplate;

	@Autowired
	private RedissonClient redissonClient;

    @Override
    @GuiGuCache(prefix = "stat:info:")
    public AlbumStatVo getAlbumStatVo(Long albumId) {
        //	获取专辑统计数据
        return albumInfoMapper.selectAlbumStatVo(albumId);
    }

    @Override
    public List<AlbumAttributeValue> findAlbumAttributeValue(Long albumId) {
        //	select * from album_attribute_value where album_id = ?;
        //	wrapper 就是在构建查询条件！
        return albumAttributeValueMapper.selectList(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId, albumId));
    }

    @Override
    public List<AlbumInfo> findUserAllAlbumList(Long userId) {
        //	select id,album_title from album_info where user_id = ? order by id desc limit 10;
        //	this.list() = albumInfoMapper.selectList();
        //		LambdaQueryWrapper<AlbumInfo> wrapper = new LambdaQueryWrapper<>();
        //		wrapper.eq(AlbumInfo::getUserId,userId);
        //		wrapper.orderByDesc(AlbumInfo::getId);
        //		wrapper.select(AlbumInfo::getId,AlbumInfo::getAlbumTitle);
        //		wrapper.last("limit 10");
        //		return albumInfoMapper.selectList(wrapper);
        //	select id,album_title from album_info where user_id = ? order by id desc limit page ,limit ;
        Page<AlbumInfo> albumInfoPage = new Page<>(1, 10);
        LambdaQueryWrapper<AlbumInfo> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(AlbumInfo::getUserId, userId);
        wrapper.orderByDesc(AlbumInfo::getId);
        wrapper.select(AlbumInfo::getId, AlbumInfo::getAlbumTitle);
        return albumInfoMapper.selectPage(albumInfoPage, wrapper).getRecords();
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void updateAlbumInfo(Long albumId, AlbumInfoVo albumInfoVo) {
        //  先删除： album:info:[1603]
        String albumKey = "album:info:["+albumId+"]";
        //  调用删除方法.
        this.redisTemplate.delete(albumKey);
        //	修改album_info
        //	第一个参数是要修改的内容 第二个参数wrapper 构建条件
        AlbumInfo albumInfo = new AlbumInfo();
        //	属性覆盖
        BeanUtils.copyProperties(albumInfoVo, albumInfo);
        this.update(albumInfo, new LambdaQueryWrapper<AlbumInfo>().eq(AlbumInfo::getId, albumId));

        //	album_attribute_value 先删除，再新增！
        this.albumAttributeValueMapper.delete(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId, albumId));
        //	新增： albumAttributeValueVoList
        List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
        //	判断
        if (!CollectionUtils.isEmpty(albumAttributeValueVoList)) {
            List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueVoList.stream().map(albumAttributeValueVo -> {
                //	创建对象
                AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
                //	赋值：如果 albumAttributeValueVo 这个实体类中要是有 albumId = null; 那么这个albumAttributeValue.setAlbumId(albumId); 应该写在属性拷贝之后
                albumAttributeValue.setAlbumId(albumId);
                //	属性拷贝：
                BeanUtils.copyProperties(albumAttributeValueVo, albumAttributeValue);
                //	返回数据
                return albumAttributeValue;
            }).collect(Collectors.toList());
            //	保存属性数据
            albumAttributeValueService.saveBatch(albumAttributeValueList);
        }

        //  睡眠
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        //  调用删除方法.
        this.redisTemplate.delete(albumKey);
        //	判断是否判断是否上架
        if ("1".equals(albumInfoVo.getIsOpen())) {
            //	发送一个消息： 发送内容是根据消费者来决定的！上架
            this.kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_UPPER, albumId.toString());
        } else {
            //	下架
            this.kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_LOWER, albumId.toString());
        }
    }

    @Override
    @GuiGuCache(prefix = RedisConstant.ALBUM_INFO_PREFIX)
    public AlbumInfo getAlbumInfo(Long albumId) {
        //  只查询数据.
        return getAlbumInfoById(albumId);
        //  return getAlbumInfoTryLock(albumId);
        //  return getAlbumInfoLock(albumId);
        //	return getAlbumInfoByRedis(albumId);
    }

    @Nullable
    private AlbumInfo getAlbumInfoTryLock(Long albumId) {
        //	先查询缓存：
        String albumKey = RedisConstant.ALBUM_INFO_PREFIX + albumId;
        //	根据key 来获取数据
        try {
            AlbumInfo albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(albumKey);
            if (null == albumInfo) {
                //	声明锁的key
                String albumLocKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId;
                RLock lock = this.redissonClient.getLock(albumLocKey);
                //	上锁。 tryLock();
                boolean result = false;
                try {
                    result = lock.tryLock(RedisConstant.ALBUM_LOCK_EXPIRE_PX1, RedisConstant.ALBUM_LOCK_EXPIRE_PX2, TimeUnit.SECONDS);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                //  获取到锁
                if (result){
                    try {
                        System.out.println("获取都分布式锁.");
                        //	获取到锁.
                        //	说明缓存中没有数据.
                        albumInfo = getAlbumInfoById(albumId);
                        //	判断能否从数据库中查询到数据.
                        if (null == albumInfo) {
                            //	声明一个空对象
                            albumInfo = new AlbumInfo();
                            //	存储数据
                            this.redisTemplate.opsForValue().set(albumKey, albumInfo, RedisConstant.ALBUM_TEMPORARY_TIMEOUT, TimeUnit.SECONDS);
                            //	返回数据
                            return albumInfo;
                        }
                        //	将数据存储到缓存中。
                        this.redisTemplate.opsForValue().set(albumKey, albumInfo, RedisConstant.ALBUM_TIMEOUT, TimeUnit.SECONDS);
                        //	返回数据
                        return albumInfo;
                    } finally {
                        lock.unlock();
                    }
                } else {
                    //  Thread.sleep(100);
                    return getAlbumInfo(albumId);
                }

            } else {
                System.out.println("走缓存...");
                return albumInfo;
            }
        } catch (RuntimeException e) {
            //	打印日志信息.
            log.error("出现异常了：" + e.getMessage());
            //	调用阿里短信-电话api 接口给工程师打电话发短信！
        }
        //	数据库扛住
        return this.getAlbumInfoById(albumId);
    }

    @Nullable
    private AlbumInfo getAlbumInfoLock(Long albumId) {
        //	先查询缓存：
        String albumKey = RedisConstant.ALBUM_INFO_PREFIX + albumId;
        //	根据key 来获取数据
        //	this.redisTemplate.opsForValue().set(albumKey,AlbumInfo);
        try {
            AlbumInfo albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(albumKey);
            if (null == albumInfo) {
                //	声明锁的key
                String albumLocKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId;
				RLock lock = this.redissonClient.getLock(albumLocKey);
				//	上锁。 tryLock();
				lock.lock();
				try {
					System.out.println("获取都分布式锁.");
					//	要再次查询数据。
					albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(albumKey);
					if (null != albumInfo) {
                        return albumInfo;
					}
					//	获取到锁.
					//	说明缓存中没有数据.
					albumInfo = getAlbumInfoById(albumId);
					//	判断能否从数据库中查询到数据.
					if (null == albumInfo) {
						//	声明一个空对象
						albumInfo = new AlbumInfo();
						//	存储数据
						this.redisTemplate.opsForValue().set(albumKey, albumInfo, RedisConstant.ALBUM_TEMPORARY_TIMEOUT, TimeUnit.SECONDS);
						//	返回数据
                        return albumInfo;
					}
					//	将数据存储到缓存中。
					this.redisTemplate.opsForValue().set(albumKey, albumInfo, RedisConstant.ALBUM_TIMEOUT, TimeUnit.SECONDS);
					//	返回数据
                    return albumInfo;
				} finally {
					lock.unlock();
				}
            } else {
                System.out.println("走缓存...");
                return albumInfo;
            }
        } catch (RuntimeException e) {
            //	打印日志信息.
            log.error("出现异常了：" + e.getMessage());
            //	调用阿里短信-电话api 接口给工程师打电话发短信！
        }
        //	数据库扛住
        return this.getAlbumInfoById(albumId);
    }

    @Nullable
    private AlbumInfo getAlbumInfoByRedis(Long albumId) {
        //	先查询缓存：
        String albumKey = RedisConstant.ALBUM_INFO_PREFIX + albumId;
        //	根据key 来获取数据
        //	this.redisTemplate.opsForValue().set(albumKey,AlbumInfo);
        try {
            AlbumInfo albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(albumKey);
            if (null == albumInfo) {
                //	声明锁的key
                String albumLocKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId;
                String token = UUID.randomUUID().toString().replaceAll("-", "");
                //	redis+lua;
                Boolean result = this.redisTemplate.opsForValue().setIfAbsent(albumLocKey, token, RedisConstant.ALBUM_LOCK_EXPIRE_PX2, TimeUnit.SECONDS);
                //	判断是否上锁成功.
                if (result) {
                    System.out.println("获取都分布式锁.");
                    //	获取到锁.
                    //	说明缓存中没有数据.
                    albumInfo = getAlbumInfoById(albumId);
                    //	判断能否从数据库中查询到数据.
                    if (null == albumInfo) {
                        //	声明一个空对象
                        albumInfo = new AlbumInfo();
                        //	存储数据
                        this.redisTemplate.opsForValue().set(albumKey, albumInfo, RedisConstant.ALBUM_TEMPORARY_TIMEOUT, TimeUnit.SECONDS);
                        //	删除锁
                        delLockKey(albumLocKey, token);
                        //	返回数据
                        return albumInfo;
                    }
                    //	将数据存储到缓存中。
                    this.redisTemplate.opsForValue().set(albumKey, albumInfo, RedisConstant.ALBUM_TIMEOUT, TimeUnit.SECONDS);
                    //	删除锁，需要使用lua脚本.
                    delLockKey(albumLocKey, token);
                    //	返回数据
                    return albumInfo;
                } else {
                    //	自旋：
                    try {
                        Thread.sleep(100);
                        return getAlbumInfo(albumId);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
            } else {
                System.out.println("走缓存...");
                return albumInfo;
            }
        } catch (RuntimeException e) {
            //	打印日志信息.
            log.error("出现异常了：" + e.getMessage());
            //	调用阿里短信-电话api 接口给工程师打电话发短信！
        }
        //	数据库扛住
        return this.getAlbumInfoById(albumId);
    }

    private void delLockKey(String albumLocKey, String token) {
        //  定义一个lua脚本
        String scriptText = "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
                "then\n" +
                "    return redis.call(\"del\",KEYS[1])\n" +
                "else\n" +
                "    return 0\n" +
                "end";

        //  创建RedisScript 对象
        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptText(scriptText);
        redisScript.setResultType(Long.class);
        //  第二个参数是锁的key 第三个参数：token
        this.redisTemplate.execute(redisScript, Arrays.asList(albumLocKey), token);
    }

    @Nullable
    private AlbumInfo getAlbumInfoById(Long albumId) {
        //	根据主键查询数据
        AlbumInfo albumInfo = this.getById(albumId);
        //	分类数据自动发送的请求
        //	回显专辑标签属性数据：
        if (null != albumInfo) {
            //	根据专辑Id 获取集合数据
            //	构建条件
            LambdaQueryWrapper<AlbumAttributeValue> wrapper = new LambdaQueryWrapper<>();
            wrapper.eq(AlbumAttributeValue::getAlbumId, albumId);
            List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueMapper.selectList(wrapper);
            //	赋值：
            albumInfo.setAlbumAttributeValueVoList(albumAttributeValueList);
        }
        //	返回数据
        return albumInfo;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void removeAlbumInfoById(Long albumId) {
        //	album_info 表； album_stat 统计信息 album_attribute_value:
        albumInfoMapper.deleteById(albumId);
        //	this=AlbumInfoService albumInfoService = new AlbumInfoServiceImpl();
        //	this.removeById(albumId);
        //	删除对应的专辑统计信息，不能使用ById!
        //	Wrapper ： 就是构建sql 语句中的条件!
        //	delete from album_stat where album_id = ? 物理删除： 逻辑删除本质是update 语句
        //	QueryWrapper:
        //  QueryWrapper<AlbumStat> wrapper = new QueryWrapper<>();
        //  wrapper.eq("album_id",albumId);
        LambdaQueryWrapper<AlbumStat> wrapper = new LambdaQueryWrapper<>();
        wrapper.eq(AlbumStat::getAlbumId, albumId);
        albumStatMapper.delete(wrapper);

        //	删除数据：album_attribute_value
        LambdaQueryWrapper<AlbumAttributeValue> wrapper1 = new LambdaQueryWrapper<>();
        wrapper1.eq(AlbumAttributeValue::getAlbumId, albumId);
        albumAttributeValueMapper.delete(wrapper1);

        //下架
        kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_LOWER, String.valueOf(albumId));
    }

    @Override
    public IPage<AlbumListVo> findUserAlbumPage(Page<AlbumListVo> albumListVoPage, AlbumInfoQuery albumInfoQuery) {
        //	返回数据
        return albumInfoMapper.selectUserAlbumPage(albumListVoPage, albumInfoQuery);
    }

    @Override
    public IPage<AlbumListVo> findUserAlbumPage(Page<AlbumListVo> albumListVoPage, Long userId, AlbumInfoQuery albumInfoQuery) {
        //	返回数据
        return albumInfoMapper.selectUserAlbumPage(albumListVoPage, userId, albumInfoQuery);
    }

    @Override
    @Transactional(rollbackFor = Exception.class) // 本地事务-声明式事务 底层使用的思想-aop 反射
    public void saveAlbumInfo(AlbumInfoVo albumInfoVo, Long userId) {
        //		album_info
        //	声明对象
        AlbumInfo albumInfo = new AlbumInfo();
        //	赋值：
        //	albumInfo.setUserId(userId);
        //	给个默认值1
        albumInfo.setUserId(1l);
        //	属性拷贝：
        BeanUtils.copyProperties(albumInfoVo, albumInfo);
        //	表示默认审核通过
        albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
        //	判断当前专辑是否付费
        if (!albumInfo.getPayType().equals(SystemConstant.ALBUM_PAY_TYPE_FREE)) {
            //	设置前5集免费
            albumInfo.setTracksForFree(5);
        }
        //	保存专辑
        albumInfoMapper.insert(albumInfo);
        //	打印数据
        System.out.println("list:" + albumInfo.getAlbumAttributeValueVoList());
        //	获取到属性集合 album_attribute_value
        List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
        if (!CollectionUtils.isEmpty(albumAttributeValueVoList)) {
            //	循环遍历
            //			albumAttributeValueVoList.forEach(albumAttributeValueVo -> {
            //				//	创建属性标签对象
            //				AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
            //				//	赋值：album_id 前端没有传递
            //				//	因为在这段代码之前已经执行了insert 语句。  @TableId(type = IdType.AUTO)
            //				//	因为这个IdType.AUTO 就能够获取到当前的主键Id 了
            //				albumAttributeValue.setAlbumId(albumInfo.getId());
            //				BeanUtils.copyProperties(albumAttributeValueVo,albumAttributeValue);
            //				//	保存数据 执行5条 insert 语句
            //				albumAttributeValueMapper.insert(albumAttributeValue);
            //			});
            List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueVoList.stream().map(albumAttributeValueVo -> {
                //	创建属性标签对象
                AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
                //	赋值：album_id 前端没有传递
                //	因为在这段代码之前已经执行了insert 语句。  @TableId(type = IdType.AUTO)
                //	因为这个IdType.AUTO 就能够获取到当前的主键Id 了
                albumAttributeValue.setAlbumId(albumInfo.getId());
                BeanUtils.copyProperties(albumAttributeValueVo, albumAttributeValue);
                //	返回
                return albumAttributeValue;
            }).collect(Collectors.toList());

            //	批量插入数据：insert into album_attribute_value values(?,?,?)
            albumAttributeValueService.saveBatch(albumAttributeValueList);
        }

        //	album_stat 统计信息初始化
        this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_PLAY);
        this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_SUBSCRIBE);
        this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_BROWSE);
        this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_COMMENT);

        //	判断是否判断是否上架
        if ("1".equals(albumInfoVo.getIsOpen())) {
            //	发送一个消息： 发送内容是根据消费者来决定的！
            this.kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_UPPER, albumInfo.getId().toString());
        }

    }

    /**
     * 保存专辑统计信息
     *
     * @param albumId
     * @param statType
     */
    private void saveAlbumStat(Long albumId, String statType) {
        //	创建实体类
        AlbumStat albumStat = new AlbumStat();
        //	赋值
        albumStat.setAlbumId(albumId);
        albumStat.setStatType(statType);
        albumStat.setStatNum(0);
        //	保存信息
        albumStatMapper.insert(albumStat);
    }
}
